package cn.itcast.flink.trans;

import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Author itcast
 * Date 2022/1/13 9:26
 * Desc 对流中的数据按照奇数和偶数进行分流，并获取分流后的数据。
 * 新版本中Flink的分流操作不再使用 split ，而是使用 process()方法
 * 将满足条件的数据 通过 OutputTag 侧输出流进行收集操作
 * 侧输出流分出来的管道
 */
public class SplitStreamDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //获取一个数据序列 1~20
        DataStreamSource<Long> source = env.fromSequence(1, 20);
        //定义侧输出流用于收集 奇数或者偶数
        OutputTag<Long> even = new OutputTag("even", Types.LONG());
        //奇数
        OutputTag<Long> odd = new OutputTag("odd", Types.LONG());
        //对其进行分流操作，将奇数放到一个数据流中，将偶数放到另外一个数据流中
        //数据流.process(new ProcessFunction(){})
        // 输入的类型是 Long 类型，输出的奇数或偶数还是 Long 类型
        SingleOutputStreamOperator<Long> result = source.process(new ProcessFunction<Long, Long>() {
            @Override
            public void processElement(Long value, Context ctx, Collector<Long> out) throws Exception {
                if (value % 2 == 0) {
                    //System.out.println("偶数 even");
                    ctx.output(even, value);
                } else {
                    ctx.output(odd, value);
                    //out.collect(value);
                }
            }
        });

        //打印输出奇数流和偶数流
        result.print("主干：|");
        result.getSideOutput(even).print("even");
        result.getSideOutput(odd).print("odd");
        //执行流环境
        env.execute();
    }
}
